package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo8ReduceByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("reduceByKey")

    val sc = new SparkContext(conf)

    //分数表
    val linesRDD: RDD[String] = sc.textFile("data/score.txt")

    /**
     * 统计学生的总分
     *
     */

    val scoreRDD: RDD[(String, Int)] = linesRDD.map((line: String) => {
      val split: Array[String] = line.split(",")
      (split.head, split.last.toInt)
    })

    /**
     * reduceByKey: 按照key对value做聚合计算，
     * 会在map端做预聚合，可以建设shuffle过程中需要传输的数据量，可以提高计算的效率
     * 1、现在map端将同一个key的value聚合在一起
     * 2、再到reduce端全局聚合
     *
     * reduceByKey： 只能用于聚合计算，不能用于分组之后处理复杂逻辑
     *
     */

    val sumScoreRDD: RDD[(String, Int)] = scoreRDD.reduceByKey((x: Int, y: Int) => {
      println(s"x + y = $x + $y")
      x + y
    })


    sumScoreRDD.foreach(println)

    //简写
    //scoreRDD.reduceByKey((x: Int, y: Int) => x + y)
    scoreRDD.reduceByKey(_ + _).foreach(println)


    while (true) {}
  }

}
